package kafka

import (
	"context"
	"errors"
	"sync"
	"time"

	"github.com/google/uuid"

	kafka "github.com/segmentio/kafka-go"
	"testkjcld.bhecard.com/gitlab/kit/esim/config"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
)

type Writer struct {
	writer *kafka.Writer

	compression Compression

	brokers []string

	logger log.Logger

	conf config.Config

	clientID string

	batchSize int

	batchTimeout time.Duration
}

type (
	WriterOption func(*Writer)
	Header       = kafka.Header
	Compression  = kafka.Compression
)

var (
	writeOne sync.Once
	writer   *Writer
)

func NewWriter(options ...WriterOption) *Writer {
	writeOne.Do(func() {
		p := &Writer{}

		for _, option := range options {
			option(p)
		}

		if p.conf == nil {
			p.conf = config.NewNullConfig()
		}

		if p.logger == nil {
			p.logger = log.NewLogger()
		}

		p.brokers = p.conf.GetStringSlice("kafka_brokers")
		if p.brokers == nil {
			p.logger.Panicf("kafka_brokers is null! please confirm")
		}

		if p.clientID == "" {
			p.clientID = p.conf.GetString("kafka_client_id")
		}

		if p.batchSize == 0 {
			p.batchSize = 1
		}

		if p.batchTimeout.Milliseconds() == 0 {
			p.batchTimeout = 100 * time.Millisecond
		}

		compString := p.conf.GetString("kafka_compression")
		switch compString {
		case "gzip":
			p.compression = kafka.Gzip
		case "snappy":
			p.compression = kafka.Snappy
		case "lz4":
			p.compression = kafka.Lz4
		case "zstd":
			p.compression = kafka.Zstd
		default: // no compression
			p.compression = 0
		}

		/*初始化生产者*/
		// p.writer = &kafka.Writer{
		// 	Addr:         kafka.TCP(p.brokers...),
		// 	Balancer:     &kafka.LeastBytes{},
		// 	Compression:  p.compression,
		// 	BatchTimeout: 100 * time.Millisecond, // 100ms 写一次
		// 	BatchSize:    1,                      // 有1条就发送
		// 	Logger:       p.logger,
		// 	RequiredAcks: kafka.RequireOne,
		// }

		p.writer = kafka.NewWriter(kafka.WriterConfig{
			Brokers:          p.brokers,
			Balancer:         &kafka.LeastBytes{},
			CompressionCodec: p.compression.Codec(),
			BatchTimeout:     p.batchTimeout,
			BatchSize:        p.batchSize,
			Logger:           p.logger,
			RequiredAcks:     int(kafka.RequireOne),
			Dialer: &kafka.Dialer{
				ClientID:  p.clientID,
				Timeout:   10 * time.Second,
				DualStack: true,
			},
		})

		writer = p
	})

	return writer
}

func WithWriterConf(conf config.Config) WriterOption {
	return func(p *Writer) {
		p.conf = conf
	}
}

func WithWriterLogger(logger log.Logger) WriterOption {
	return func(p *Writer) {
		p.logger = logger
	}
}

func WithWriterClientID(clientID string) WriterOption {
	return func(p *Writer) {
		p.clientID = clientID
	}
}

func WithWriterBatchTimeout(batchTimeout time.Duration) WriterOption {
	return func(p *Writer) {
		p.writer.BatchTimeout = batchTimeout
	}
}

func WithWriterBatchSize(batchSize int) WriterOption {
	return func(p *Writer) {
		p.writer.BatchSize = batchSize
	}
}

func (p *Writer) PublishMessage(ctx context.Context, topicName, messageBody string, header []Header) error {
	if topicName == "" {
		return errors.New("topicName is null")
	}
	msg := kafka.Message{
		Topic:   topicName,
		Headers: header,
		Key:     msgId(),
		Value:   []byte(messageBody),
	}
	return p.write(ctx, msg)
}

// WriteMessage 返回msg 信息
func (p *Writer) WriteMessage(ctx context.Context, topicName, messageBody string, header []Header) (*kafka.Message, error) {
	if topicName == "" {
		return nil, errors.New("topicName is null")
	}
	msg := kafka.Message{
		Topic:   topicName,
		Headers: header,
		Key:     msgId(),
		Value:   []byte(messageBody),
	}
	return &msg, p.write(ctx, msg)
}

func (p *Writer) PublishKeyMessage(ctx context.Context, topicName string, messageKey, messageBody []byte, header []Header) error {
	if topicName == "" {
		return errors.New("topicName is null")
	}
	msg := kafka.Message{
		Topic:   topicName,
		Key:     messageKey,
		Value:   messageBody,
		Headers: header,
	}
	return p.write(ctx, msg)
}

func (p *Writer) write(ctx context.Context, msg kafka.Message) error {
	err := p.writer.WriteMessages(ctx, msg)
	if err != nil {
		p.logger.Errorc(ctx, "WriteMessages失败[%s]:[%s]", msg.Topic, err)
		return err
	}
	return nil
}

func msgId() []byte {
	uid, _ := uuid.NewUUID()
	return []byte(uid.String())
}
